package com.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo7Window {
  def main(args: Array[String]): Unit = {

    /**
     * 窗口函数
     * row_number
     * rank
     * sum
     * count
     * avg
     * max
     * min
     * lag 取当前行前面
     * lead： 取当前行后面的
     *
     */

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("window")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    import org.apache.spark.sql.functions._
    import spark.implicits._

    //学生表
    val studentDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
      .load("data/students.txt")

    //分数表
    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("sid STRING,cid STRING,sco DOUBLE")
      .load("data/score.txt")

    //科目表
    val subjectDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("cid STRING,cname STRING,ssco DOUBLE")
      .load("data/subject.txt")


    val joinDF: DataFrame = studentDF.join(scoreDF, $"id" === $"sid")

    /**
     * 1、统计总分年级排名前十学生各科的分数
     *
     * sum over 由两种用法
     * 1、分区不排序----总和
     * 2、分区也排序-----累计求和
     */

    joinDF
      //计算总分
      .withColumn("sumSco", sum($"sco") over Window.partitionBy($"id"))
      //按总分排名
      .orderBy($"sumSco".desc)
      //取top
      .limit(60)
    //.show(60)

    /**
     * 3、统计每科都及格的学生
     * 判断学生的成绩得分是否大于等于总分 * 0.6
     */
    scoreDF
      //关联科目表
      .join(subjectDF, "cid")
      //1过滤不及格的分数
      .where($"sco" >= $"ssco" * 0.6)
      //统计每个学生几个的科目数
      .withColumn("jige", count($"sid") over Window.partitionBy($"sid"))
      //取出都及格的学生
      .where($"jige" === 6)
    //.show(100)

    /**
     * 2、统计总分大于年级平均分的学生
     *
     */

    joinDF
      //计算总分
      .withColumn("sumSco", sum($"sco") over Window.partitionBy($"id"))
      //计算年级平均分
      .withColumn("avgSco", avg($"sumSco") over Window.partitionBy(substring($"clazz", 0, 2)))
      .where($"sumSco" > $"avgSco")
    //.show(1000)


    /**
     * 统计每个班级每个学生名次分数差
     *
     * lag:取当前行前面的数据，需要分区和排序
     * lead 取当前行后面的数据，需要分区和排序
     */

    joinDF
      //按照学号和班级分组
      .groupBy($"id", $"clazz")
      //计算总分
      .agg(sum($"sco") as "sumSco")
      // 增加排名
      // .withColumn("r", row_number() over Window.partitionBy($"clazz").orderBy($"sumSco".desc))
      // 取前一个名次的总分
      .withColumn("headSumSco", lag($"sumSco", 1, 750)
        over Window.partitionBy($"clazz").orderBy($"sumSco".desc))
      // 计算分数差
      .withColumn("cha", $"headSumSco" - $"sumSco")
      .show(100)

  }
}
